package com.qf.flashlib.util;


import com.qf.flashlib.base.BaseResponse;
import com.qf.flashlib.exception.ApiException;
import com.qf.flashlib.exception.ServerException;
import com.qf.flashlib.http.LoggerUtil;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * 2018/11/9 15:41
 */
public class RxUtils {
    private RxUtils() {}

    /**
     * 线程调度
     */
    public static <T> ObservableTransformer<T, T> io_main() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
                return upstream
                        .subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(@NonNull Disposable disposable) throws Exception {
                                LoggerUtil.i("+++doOnSubscribe+++" + disposable.isDisposed());
                            }
                        })
                        .doFinally(new Action() {
                            @Override
                            public void run() throws Exception {
                                LoggerUtil.i("+++doFinally+++");
                            }
                        })
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    /**
     * 结果处理
     */
    public static <T> ObservableTransformer<BaseResponse<T>,T> handleResult() {
        return new ObservableTransformer<BaseResponse<T>, T>() {
            @Override
            public ObservableSource<T> apply(Observable<BaseResponse<T>> upstream) {
               return upstream.flatMap(new Function<BaseResponse<T>, ObservableSource<T>>() {
                    @Override
                    public Observable<T> apply(final BaseResponse<T> tBaseResponse) throws Exception {
                        if (tBaseResponse.isOk()) {
                            return Observable.create(new ObservableOnSubscribe<T>() {
                                @Override
                                public void subscribe(ObservableEmitter<T> e) throws Exception {
                                    e.onNext(tBaseResponse.getData());
                                }
                            });
                        } else {
                            return Observable.error(new ServerException(tBaseResponse.getCode(),tBaseResponse.getMsg())) ;
                        }
                    }
                }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends T>>() {
                   @Override
                   public ObservableSource<? extends T> apply(Throwable throwable) throws Exception {
                       return Observable.error(ApiException.handleException(throwable));
                   }
               });

            }
        };
    }
    /**
     * 结果处理
     */
    public static <T> ObservableTransformer<BaseResponse<T>,BaseResponse<T>> handleBaseResult() {
        return new ObservableTransformer<BaseResponse<T>,BaseResponse<T>>() {
            @Override
            public ObservableSource<BaseResponse<T>> apply(Observable<BaseResponse<T>> upstream) {
               return upstream.flatMap(new Function<BaseResponse<T>, ObservableSource<BaseResponse<T>>>() {
                    @Override
                    public Observable<BaseResponse<T>> apply(final BaseResponse<T> tBaseResponse) throws Exception {
                        if (tBaseResponse.isOk()) {
                            return Observable.create(new ObservableOnSubscribe<BaseResponse<T>>() {
                                @Override
                                public void subscribe(ObservableEmitter<BaseResponse<T>> e) throws Exception {
                                    e.onNext(tBaseResponse);
                                }
                            });
                        } else {
                            return Observable.error(new ServerException(tBaseResponse.getCode(),tBaseResponse.getMsg()));
                        }
                    }
                }).onErrorResumeNext(new Function<Throwable, ObservableSource<BaseResponse<T>>>() {
                   @Override
                   public ObservableSource<BaseResponse<T>> apply(Throwable throwable) throws Exception {
                       return Observable.error(ApiException.handleException(throwable));
                   }
               });

            }
        };
    }


}
